package cn.ilikes.tools.nosql.jpa.clients.indexs.index.hadoops;

import java.io.IOException;

import org.apache.lucene.store.Lock;
import org.apache.lucene.store.LockFactory;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.SysConfigs;
import cn.ilikes.tools.nosql.jpa.configs.indexs.IndexConfigs;
import cn.ilikes.tools.unilts.SQLUilts;

public class HadoopZooKeepLockFactory extends LockFactory {

	public static String lock_Id = "lock_" + SQLUilts.generatorUUID();

	private ZooKeeper zooKeeper;

	private IndexConfigs indexConfigs;

	private String lockNode;

	private Logger logger = Loggerfactory.instance(HadoopZooKeepLockFactory.class);

	public HadoopZooKeepLockFactory(IndexConfigs indexConfigs) {
		this.indexConfigs = indexConfigs;
		try {
			init(indexConfigs);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void init(IndexConfigs indexConfigs) throws Exception {
		Loggerfactory.info(logger, "init ilikes  Hadoop ZooKeep LockFactory ...");
		lockNode = indexConfigs.getProperties().getProperty(SysConfigs.index_lunece_lock_node);
		String zookeeperHost=indexConfigs.getProperties().getProperty(SysConfigs.index_zookeeper_host_address);
		ZooKeeper zooKeeper = new ZooKeeper(zookeeperHost, 10 * 1000, new Watcher() {
			public void process(WatchedEvent event) {
				Loggerfactory.info(logger, "event " + event.getType() + " has happened!");
			}
		});
		setZooKeeper(zooKeeper);

	}

	private void setZooKeeper(ZooKeeper zooKeeper) {
		this.zooKeeper = zooKeeper;
	}

	@Override
	public void clearLock(String lockName) throws IOException {
		try {
			if(Loggerfactory.isdebug(logger))Loggerfactory.debug(logger, "clear  Lock  lockName :  "+lockName);
			 HadoopZooKeepLockUnilts.release(zooKeeper, lockNode,lock_Id);
		          } catch (Exception e) {
			throw new IOException(e);
		}
	}

	@Override
	public Lock makeLock(String lockName) {
		if(Loggerfactory.isdebug(logger))Loggerfactory.debug(logger, "make Lock  lockName :  "+lockName);
		HadoopZooKeepLock lock = new HadoopZooKeepLock(zooKeeper, lockNode, lock_Id);
		return lock;
	}

}
